package com.woniuedu.search.jobhandler;


import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.woniuedu.search.entity.MessageStatusEnum;
import com.woniuedu.search.entity.Messages;
import com.woniuedu.search.mapper.MessageMapper;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.handler.annotation.XxlJob;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.List;
import java.util.concurrent.TimeUnit;
@Component
public class XxlJobHandler {

    private Logger logger = LoggerFactory.getLogger(XxlJobHandler.class);

    /**
     * 简单任务(Bean模式)
     * @param param
     * @return
     * @throws Exception
     */
    @XxlJob("jobHandler2")
    public ReturnT<String> jobHandler(String param) throws Exception {
        logger.info("XXL-JOB, Hello World");
        for (int i = 0; i < 5; i++) {
            logger.info("beat at:" + i);
            TimeUnit.SECONDS.sleep(2);
        }
        return ReturnT.SUCCESS;
    }


    @Resource
    private RabbitTemplate rabbitTemplate;

    @Autowired(required = false)
    private MessageMapper messageMapper;


    /**
     * 最大重试次数
     */
    @Value("${app.rabbitmq.retry.max-retry-times}")
    private int retryTimes;

    /**
     * 重试时间间隔
     */
    @Value("${app.rabbitmq.retry.retry-interval}")
    private Duration retryInterval;

    /**
     * 批量从数据库中读取的消息
     */
    private static final int PAGE_SIZE = 100;

    /**
     * 生命周期任务示例：任务初始化与销毁时，支持自定义相关逻辑
     * @param param
     * @return
     * @throws Exception
     */
    @XxlJob(value = "jobHandler", init = "init", destroy = "destroy")
    public ReturnT<String> jobHandler2(String param) throws Exception {
        long startTime = System.currentTimeMillis();
        logger.info("开始扫描需要进行重试投递的消息");
        service(1);
        logger.info("扫描需要进行重试投递的消息任务结束，耗时[{}]ms", System.currentTimeMillis() - startTime);
        return ReturnT.SUCCESS;
    }


    public void service(int pageNum) {
        IPage<Messages> messageIPage = messageMapper.selectPage(new Page<>(pageNum, PAGE_SIZE),
                Wrappers.<Messages>lambdaQuery()
                        //发送中的消息
                        .eq(com.woniuedu.search.entity.Messages::getStatus, MessageStatusEnum.SENDING.getStatus())
                        //已到达下次发送时间
                        .le(com.woniuedu.search.entity.Messages::getNextRetryDateTime, LocalDateTime.now(ZoneOffset.ofHours(8)))
        );
        List<Messages> messages = messageIPage.getRecords();
        messages.forEach(message -> {
            if (retryTimes <= message.getRetryTimes()) {
                //已达到最大投递次数，将消息设置为投递失败
                messageMapper.update(null, Wrappers.<Messages>lambdaUpdate().set(com.woniuedu.search.entity.Messages::getStatus, MessageStatusEnum.FAIL.getStatus()).eq(com.woniuedu.search.entity.Messages::getId, message.getId()));
            } else {
                reSend(message);
            }
        });
        if (PAGE_SIZE == messages.size()) {
            service(++pageNum);
        }
    }


    /**
     * 重新投递消息
     *
     * @param messages
     */
    public void reSend(Messages messages) {
        messageMapper.update(null,
                Wrappers.<Messages>lambdaUpdate()
                        .set(com.woniuedu.search.entity.Messages::getRetryTimes, messages.getRetryTimes() + 1)
                        .set(com.woniuedu.search.entity.Messages::getNextRetryDateTime, LocalDateTime.now(ZoneOffset.ofHours(8)).plus(retryInterval))
                        .eq(com.woniuedu.search.entity.Messages::getId, messages.getId())
        );
        try {
            //再次投递
            rabbitTemplate.convertAndSend(messages.getExchangeName(), messages.getRoutingKey(), messages.getMsgData(), new CorrelationData(messages.getId().toString()));
        } catch (Exception e) {
            logger.error("消息[{}]投递失败", JSON.toJSONString(messages));
        }
    }
    public void init() {
        logger.info("init");
    }

    public void destroy() {
        logger.info("destory");
    }
}
